package com.lx.boot.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.lx.boot.web.valid.Validate;
import com.lx.util.LX;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Executor;

@Slf4j
@Configuration
@ConditionalOnClass(CanalConnector.class)
@EnableConfigurationProperties(CanalProperties.class)
public class CanalConfiguration implements CommandLineRunner {

    @Autowired
    private Executor executor;

    @Autowired
    private CanalProperties canalProperties;

    @Autowired
    private List<CanalService> canalServices;

    private final Map<String, List<CanalService>> tables = new ConcurrentHashMap<>();

    private final List<CanalService> allTables = new ArrayList<>();

//    # 数据库创建用户授权
//    create user 'canal'@'%' identified by 'canal@LL@123';
//    grant all privileges on *.* to 'canal'@'%';
//    flush privileges;
    //    docker run --name canal -e canal.instance.master.address=172.31.0.4:3306 -e canal.instance.dbUsername=canal -e canal.instance.dbPassword=canal@LL@123 -p 11111:11111 -d canal/canal-server:v1.1.6
    @Override
    public void run(String... args) throws Exception {
        if (LX.isEmpty(canalServices) || !canalProperties.isEnable()){
            return;
        }
        StringBuilder subscribe = new StringBuilder();
        canalServices.forEach(canalService -> {
            if (LX.isNotEmpty(canalService.getTableName())){
                List<CanalService> set = tables.get(canalService.getTableName());
                if (set == null){
                    tables.put(canalService.getTableName(),Arrays.asList(canalService));
                }else{
                    set.add(canalService);
                }
            }else{
                allTables.add(canalService);
            }
        });
        Thread thread = new Thread(() -> {
            while (true) {
                connection();
            }
        });
        thread.setName("canalUtilThread");
        thread.start();
    }

    private void connection() {
        Validate.validate(canalProperties);
        //连接Canal服务器
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalProperties.getIp(), canalProperties.getProt())
                , canalProperties.getDestination(), canalProperties.getUsername(), canalProperties.getPassword());
        try {
            connector.connect();
            //监听的表 格式为数据库.表名,数据库.表名  全部:  .*.*
            connector.subscribe(canalProperties.getSubscribe());
            connector.rollback();
            while (true) {
                long batchId = 0;
                try {
                    // 获取指定数量的数据
                    Message message = connector.getWithoutAck(1);
                    batchId = message.getId();
                    if (batchId == -1 || message.getEntries().isEmpty()) {
                        LX.sleep(3000);
                        continue;
                    }
                    exec(message.getEntries());
                    // 提交确认，消费成功，通知server删除数据
                    connector.ack(batchId);
                }catch (Exception e1){
                    log.error("",e1);
                    // 处理失败, 回滚数据，后续重新获取数据
                    connector.rollback(batchId);
                    LX.sleep(3000);
                }
            }
        }catch (Exception e){
            log.error("连接canal失败!",e);
            LX.sleep(10000);
        }finally {
            connector.disconnect();
        }
    }

    private void exec(List<Entry> entries) throws Exception {
        for (Entry entry : entries) {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }
            String tableName = entry.getHeader().getTableName();
            List<CanalService> services = tables.get(tableName);
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            CanalEntry.EventType eventType = rowChange.getEventType();
            log.info(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));
            if (services != null){
                for (CanalService service : services) {
                    service.notifiy(tableName,eventType, rowChange);
                }
            }
            for (CanalService service : allTables) {
                service.notifiy(tableName,eventType, rowChange);
            }
        }

    }


}
